﻿using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.IO;
using System.Text;
using System.Threading;
using Common.Utilities;
using Common.Utilities.Dispatcher;
using Common.Utilities.OpenXml;

namespace Workflows.Components.ETL
{
	/// <summary>
	/// 
	/// </summary>
	public class DelimitedTextFileReaderWorker:IWorker 
	{
		#region Implementation of IWorker

		public event WorkerStarted WorkerStartedEvent;
		public event WorkerFinished WorkerFinishedEvent;
		public event WorkerHasError WorkerErredEvent;
		public event WorkerCancelled WorkerCancelledEvent;
		public event WorkerStatusChanged WorkerStatusChangedEvent;

		private Thread workerThread;
		private DelimitedTextFileReaderInstruction instruction;
		private DelimitedTextFileReaderOutput output;
		private RunContext _Context;
		private Guid _TicketID;
		private string _WorkflowName;
		private string _ComponentName;

		/// <summary>
		/// 
		/// </summary>
		public string ComponentName
		{
			get { return _ComponentName; }
		}

		/// <summary>
		/// 
		/// </summary>
		public string WorkflowName
		{
			get { return _WorkflowName; }
		}

		/// <summary></summary>
		public Thread WorkerThread
		{
			get { return workerThread; }
		}
		/// <summary></summary>
		public InstructionBase Instruction
		{
			get { return instruction; }
		}
		/// <summary></summary>
		public ResultBase Output
		{
			get { return output; }
		}
		/// <summary></summary>
		public Type ComponentType
		{
			get { return typeof(DelimitedTextFileReaderComponent); }
		}
		/// <summary></summary>
		public RunContext Context
		{
			get { return _Context; }
			set { _Context = value; }
		}
		/// <summary></summary>
		public void SetInstruction(Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
		{
			this._TicketID = ticketGuid;
			this.instruction = (DelimitedTextFileReaderInstruction)request;
			this._WorkflowName = workflowName;
			this._ComponentName = componentName;
		}
		/// <summary></summary>
		public void SetInstruction(RunContext context, Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
		{
			this._TicketID = ticketGuid;
			this.instruction = (DelimitedTextFileReaderInstruction)request;
			this._Context = context;
			this._WorkflowName = workflowName;
			this._ComponentName = componentName;
		}
		/// <summary></summary>
		public void DoWork()
		{
			if (this.WorkerStartedEvent != null)
			{
				this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this.instruction);
			}
			this.workerThread = Thread.CurrentThread;
			this.ReadDelimitedFile();
		}
		/// <summary></summary>
		public void DoWorkAsync()
		{
			if (this.WorkerStartedEvent != null)
			{
				this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this.instruction);
			}
			this.workerThread = new Thread(new ThreadStart(this.ReadDelimitedFile));
			this.workerThread.Start();
		}
		/// <summary></summary>
		public void CancelWork()
		{
			if (this.workerThread != null && this.workerThread.IsAlive)
			{
				this.workerThread.Abort();
				this.workerThread.Join();

				if (this.WorkerCancelledEvent != null)
				{
					this.WorkerCancelledEvent(this._TicketID, this.GetType(), this.instruction);
				}
			}
		}

		#endregion

		#region work
		private void ReadDelimitedFile()
		{
            SqlConnection conn=new SqlConnection(this.instruction.ConnStr);
			StreamReader reader = null;
			try
			{
                conn.Open();
                Dictionary<string,SqlParameter> sqlParameters=new Dictionary<string, SqlParameter>();
			    DataTable dtPreview = this.Test(200);
			    SqlCommand cmd = this.PrepareInsert(
			        conn, this.instruction.ConnStr,
			        this.instruction.SchemaName, this.instruction.TableName,
                    dtPreview, this.instruction.RecreateTable, this.instruction.ClearExistingData,
                    ref sqlParameters);

                
                DataTable dt=new DataTable();
				reader=new StreamReader(this.instruction.TxtFilePath);
				int lineNum = 0;
				bool startRead = false;
			    int totalLineNum = 0;
                while(!reader.EndOfStream)
                {
                    reader.ReadLine();
                    totalLineNum++;
                }
                reader.Close();
                reader = new StreamReader(this.instruction.TxtFilePath);
				while(!reader.EndOfStream)
				{
					string lineContent = reader.ReadLine();
					lineNum++;
					if(!startRead)
					{
						if(lineNum==this.instruction.RowNumContainingColumnHeaders)
						{
							string[] colNames = this.ReadColumnHeaders(lineContent, this.instruction.ColumnDelimiter);
							if(colNames ==null || colNames.Length <1)
								throw new Exception("Unable to find column headers.");

							foreach(string colName in colNames)
							{
								dt.Columns.Add(colName, typeof (string));
							}
							startRead = true;
						}
					}
					else
					{
						string[] fieldContents = this.ReadTableContent(lineContent, this.instruction.ColumnDelimiter);
						if(fieldContents !=null && fieldContents.Length>0)
						{
							DataRow dr = dt.NewRow();
							int fieldCount = Math.Min(dt.Columns.Count, fieldContents.Length);
							for (int colIdx = 0; colIdx < fieldCount; colIdx++)
							{
								dr[colIdx] = fieldContents[colIdx];
							}
							dt.Rows.Add(dr);
						}
					}

                    if(lineNum % 100==0 && this.WorkerStatusChangedEvent !=null)
                    {
                        string msg = string.Format("Reading {0} of {1} lines from file {2}", lineNum, totalLineNum,
                                                   Path.GetFileNameWithoutExtension(this.instruction.TxtFilePath));
                        int pct = 59;
                        this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this.instruction, this._WorkflowName,
                                                      this._ComponentName, msg, pct);

                    }

                    if (dt.Rows.Count >= 100 && this.WorkerStatusChangedEvent != null)
					{
						this.SaveTableToDB(dt, cmd, sqlParameters);
                        dt.Clear();
					}
				}

				if(dt.Rows.Count>0)
				{
                    this.SaveTableToDB(dt, cmd, sqlParameters);
				}

				this.output=new DelimitedTextFileReaderOutput(this.instruction.OwnerComponentID,this._TicketID, dt.Rows.Count);

				if (this.WorkerFinishedEvent != null)
					this.WorkerFinishedEvent(this._TicketID, this.GetType(), this.instruction, this.output);
			}
			catch (Exception ex)
			{
				if (this.WorkerErredEvent != null)
					this.WorkerErredEvent(this._TicketID, this.GetType(), this.instruction, ex.Message, true);
			}
			finally
			{
				if(reader !=null)
					reader.Close();
			}
		}

		private string[] ReadColumnHeaders(string lineContent, Delimiter colDelimiter)
		{
			return this.ReadTableContent(lineContent, colDelimiter);
		}

		private string[] ReadTableContent(string lineContent, Delimiter colDelimiter)
		{
			if (colDelimiter.SpecialDelimiter == SpecialDelimiter.Other && !string.IsNullOrEmpty(colDelimiter.DelimiterPattern))
				return StringUtil.Split(lineContent, colDelimiter.DelimiterPattern);
			else
			{
				switch (colDelimiter.SpecialDelimiter)
				{
					case SpecialDelimiter.Tab:
                        return lineContent.Split(new char[] { '\t' });
					case SpecialDelimiter.Comma:
                        return lineContent.Split(new char[] { ',' });
					case SpecialDelimiter.Space:
                        return StringUtil.Split(lineContent, " ");
					case SpecialDelimiter.SemiColon:
                        return lineContent.Split(new char[] { ';' });
					case SpecialDelimiter.VerticalBar:
                        return lineContent.Split(new char[] { '|' });
					default:
                        return lineContent.Split(new char[] { '\t' });
				}
			}
		}
		#endregion

        #region save output
        private SqlCommand PrepareInsert(
            SqlConnection conn, string connStr, string schema, string tbl,
            DataTable dt, bool createNewTable, bool clearExistingData,
            ref Dictionary<string,SqlParameter> sqlParameters)
        {
            List<TableColumn> cols=new List<TableColumn>();
            Dictionary<string,int> colNameIdx=new Dictionary<string, int>();
            foreach(DataColumn col in dt.Columns)
            {
                string colName = StoredProcUtil.TidyObjectName(col.ColumnName);
                colNameIdx.Add(colName, col.Ordinal);
                TableColumn tblCol=new TableColumn();
                tblCol.ColumnName = colName;
                tblCol.DbType = SqlDbTypeUtil.ToSqlDbType(col.DataType);
                tblCol.Size = FieldDataType.GetFieldDefaultSize(tblCol.DbType);
                cols.Add(tblCol);
            }
            if(createNewTable)
            {
                DbUtil.NewTable(DataProviderType.MSSQL, connStr, schema, tbl, cols.ToArray());
            }
            if(clearExistingData)
            {
                string sql = string.Format("truncate table {0}.[{1}]", schema, tbl);
                DbUtil.ExecuteSqlCommand(connStr,sql);
            }
            string spName = "bulkInsert_" + tbl;
            string spBody = StoredProcUtil.CreateStoredProcedureStatementForInsert(
                DataProviderType.MSSQL, spName, schema, tbl, cols.ToArray());
            DbUtil.NewStoredProc(connStr, spName, spBody);
            sqlParameters = new Dictionary<string, SqlParameter>();
            foreach (string colName in colNameIdx.Keys)
            {
                sqlParameters.Add(colName, new SqlParameter("@" + colName, DBNull.Value));
            }
            SqlCommand cmd = conn.CreateCommand();
            cmd.CommandText = spName;
            cmd.CommandType = CommandType.StoredProcedure;
            return cmd;
        }

        private void SaveTableToDB(DataTable dt, SqlCommand cmd, Dictionary<string, SqlParameter> sqlParameters)
        {
            try
            {
                foreach (DataRow dr in dt.Rows)
                {
                    cmd.Parameters.Clear();
                    for(int i=0;i<dt.Columns.Count;i++)
                    {
                        string colName = StoredProcUtil.TidyObjectName(dt.Columns[i].ColumnName);
                        sqlParameters[colName].Value = dr[i];
                        if (sqlParameters[colName].Value == null)
                            sqlParameters[colName].Value = DBNull.Value;
                        cmd.Parameters.Add(sqlParameters[colName]);
                    }
                    try
                    {
                        cmd.ExecuteNonQuery();
                    }
                    catch(Exception ex1)
                    {
                        if (this.WorkerErredEvent != null)
                            this.WorkerErredEvent(this._TicketID, this.GetType(), this.instruction, ex1.ToString(), false);
                    }
                }    
            }
            catch(Exception ex)
            {
                if (this.WorkerErredEvent != null)
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this.instruction, ex.ToString(), false);
            }
        }
        #endregion

        /// <summary>
		/// 
		/// </summary>
		/// <param name="maxRecord"></param>
		public DataTable Test(int maxRecord)
		{
			DataTable dt = new DataTable();
			StreamReader reader = null;
			try
			{
			
				reader = new StreamReader(this.instruction.TxtFilePath);
				int lineNum = 0;
				bool startRead = false;
				while (!reader.EndOfStream)
				{
					string lineContent = reader.ReadLine();
					lineNum++;
					if (!startRead)
					{
						if (lineNum == this.instruction.RowNumContainingColumnHeaders)
						{
							string[] colNames = this.ReadColumnHeaders(lineContent, this.instruction.ColumnDelimiter);
							if (colNames == null || colNames.Length < 1)
								throw new Exception("Unable to find column headers.");

							foreach (string colName in colNames)
							{
								dt.Columns.Add(colName, typeof(string));
							}
							startRead = true;
						}
					}
					else
					{
						string[] fieldContents = this.ReadTableContent(lineContent, this.instruction.ColumnDelimiter);
						if (fieldContents != null && fieldContents.Length > 0)
						{
							DataRow dr = dt.NewRow();
							int fieldCount = Math.Min(dt.Columns.Count, fieldContents.Length);
							for (int colIdx = 0; colIdx < fieldCount; colIdx++)
							{
								dr[colIdx] = fieldContents[colIdx];
							}
							dt.Rows.Add(dr);

							if(dt.Rows.Count>=maxRecord)
								break;
						}
					}

					if (lineNum % 100 == 0 && this.WorkerStatusChangedEvent != null)
					{
						string msg = string.Format("Reading {0} lines from file {1}", lineNum,
												   Path.GetFileNameWithoutExtension(this.instruction.TxtFilePath));
						int pct = 59;
						this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this.instruction, this._WorkflowName,
													  this._ComponentName, msg, pct);
					}
				}
			}
			catch (Exception ex)
			{
				if (this.WorkerErredEvent != null)
					this.WorkerErredEvent(this._TicketID, this.GetType(), this.instruction, ex.Message, true);
			}
			finally
			{
				if (reader != null)
					reader.Close();
			}

			return dt;
		}
	}
}
